package org.opens.net.threadOptimize;

import io.netty.buffer.ByteBuf;
import lombok.extern.slf4j.Slf4j;
import org.opens.byteBuffer.ByteBufferUtil;
import org.opens.net.util.MyByteBufferUtil;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;

@Slf4j
public class Worker implements Runnable {

    private Thread thread;

    private Selector workerSelector;

    private String name;

    /**
     * worker启动标记
     */
    private volatile boolean started = false;

    private ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue<>();

    public Worker(String name) {
        this.name = name;
    }

    public void open() {

    }

    // 初始化线程和selector
    public void register(SocketChannel socketChannel, Object attachment) throws IOException {
        if (this.workerSelector == null) {
            this.workerSelector = Selector.open();
        }

        if (!started) {
            thread = new Thread(this, this.name);
            thread.start();

            started = true;
            log.info("worker {} 启动成功!", this.name);
        }

        queue.add(() -> {
            try {
                socketChannel.register(this.workerSelector, SelectionKey.OP_READ, attachment);
                log.info("新连接 {} 已注册至worker中", socketChannel.getRemoteAddress());
            } catch (IOException e) {
                e.printStackTrace();
            }
        });

        // 唤醒阻塞的workerSelector.select();执行
        workerSelector.wakeup();
    }

    @Override
    public void run() {
        while (true) {
            try {
                workerSelector.select();

                Runnable task = queue.poll();
                if (task != null) {
                    task.run();
                }

                Iterator<SelectionKey> keyIterator = workerSelector.selectedKeys().iterator();
                while (keyIterator.hasNext()) {
                    SelectionKey key = keyIterator.next();
                    keyIterator.remove();

                    try {
                        if (key.isReadable()) {
                            ByteBuffer buffer = (ByteBuffer) key.attachment();
                            SocketChannel socketChannel = (SocketChannel) key.channel();
                            int readSize = socketChannel.read(buffer);

                            // 如果client执行close方法关闭连接后, read方法将会返回-1
                            if (readSize == -1) {
                                key.cancel();
                                log.info("客户端主动断开连接: {}", socketChannel);
                            } else {
                                MyByteBufferUtil.split(buffer);
                                // 这样判断是因为ByteBuffer执行compact方法后, 会将position移动到上次未读数据的末尾, limit指向可以存储数据的最大值
                                // 因此, position==limit也就表示着buffer空间不足了.
                                if (buffer.position() == buffer.limit()) {
                                    ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity() * 2);
                                    // 将旧的buffer中的数据复制过来需要读取, 所以要切换buffer为读模式
                                    buffer.flip();
                                    newBuffer.put(buffer);
                                    // 将扩容的ByteBuffer与SelectionKey关联起来, 也就是与SocketChannel关联起来
                                    key.attach(newBuffer);
                                } else {
                                    // 切换读模式
                                    buffer.flip();
                                    // 显示数据
                                    log.info("woker中的 {} 发送来新消息: {}", socketChannel.getRemoteAddress(), Charset.defaultCharset().decode(buffer));
                                    // 切换写模式
                                    buffer.clear();
                                }
                            }

                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                        key.cancel();
                        log.info("远程主机强迫关闭了一个现有的连接, read事件已取消!");
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }

        }
    }

}
